package com.shujia.datasource.sink;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.List;

import com.shujia.datasource.bean.SztData;
import org.apache.flume.Channel;
import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.EventDeliveryException;
import org.apache.flume.Transaction;
import org.apache.flume.conf.Configurable;
import org.apache.flume.sink.AbstractSink;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.common.base.Preconditions;
import com.google.common.base.Throwables;
import com.google.common.collect.Lists;

public class FlumeJDBCSink extends AbstractSink implements Configurable {
    private Logger LOG = LoggerFactory.getLogger(FlumeJDBCSink.class);

    private String hostname;
    private String port;
    private String databaseName;
    private String tableName;
    private String user;
    private String password;
    private PreparedStatement preparedStatement;
    private Connection conn;
    private int batchSize;        //每次提交的批次大小

    public FlumeJDBCSink() {
        LOG.info("================================================\n");
        LOG.info("MySqlSink start...");
        LOG.info("================================================\n");
    }

    /**
     * 实现Configurable接口中的方法：可获取配置文件中的属性
     */
    public void configure(Context context) {
        hostname = context.getString("hostname");
        Preconditions.checkNotNull(hostname, "hostname must be set!!");
        port = context.getString("port");
        Preconditions.checkNotNull(port, "port must be set!!");
        databaseName = context.getString("databaseName");
        Preconditions.checkNotNull(databaseName, "databaseName must be set!!");
        tableName = context.getString("tableName");
        Preconditions.checkNotNull(tableName, "tableName must be set!!");
        user = context.getString("user");
        Preconditions.checkNotNull(user, "user must be set!!");
        password = context.getString("password");
        Preconditions.checkNotNull(password, "password must be set!!");
        batchSize = context.getInteger("batchSize", 100000);        //设置了batchSize的默认值
        Preconditions.checkNotNull(batchSize > 0, "batchSize must be a positive number!!");
    }

    /**
     * 服务启动时执行的代码，这里做准备工作
     */
    @Override
    public void start() {
        super.start();
        try {
            //调用Class.forName()方法加载驱动程序
            Class.forName("com.mysql.cj.jdbc.Driver");
        } catch (ClassNotFoundException e) {
            e.printStackTrace();
        }

        String url = "jdbc:mysql://" + hostname + ":" + port + "/" + databaseName + "?useUnicode=true&characterEncoding=UTF-8&useSSL=false";
        //调用DriverManager对象的getConnection()方法，获得一个Connection对象

        try {
            conn = DriverManager.getConnection(url, user, password);
            conn.setAutoCommit(false);
            //创建一个Statement对象
            preparedStatement = conn.prepareStatement("insert into " + tableName + " values (?,?,?,?,?,?,?,?,?,?,?)");

        } catch (SQLException e) {
            e.printStackTrace();
            System.exit(1);
        }

    }

    /**
     * 服务关闭时执行
     */
    @Override
    public void stop() {
        super.stop();
        if (preparedStatement != null) {
            try {
                preparedStatement.close();
            } catch (SQLException e) {
                e.printStackTrace();
            }
        }

        if (conn != null) {
            try {
                conn.close();
            } catch (SQLException e) {
                e.printStackTrace();
            }
        }
    }

    /**
     * 执行的事情：<br/>
     * （1）持续不断的从channel中获取event放到batchSize大小的数组中<br/>
     * （2）event可以获取到则进行event处理，否则返回Status.BACKOFF标识没有数据提交<br/>
     * （3）batchSize中有内容则进行jdbc提交<br/>
     */
    public Status process() throws EventDeliveryException {
        Status result = Status.READY;
        Channel channel = getChannel();
        Transaction transaction = channel.getTransaction();
        Event event;
        String content;

        List<SztData> sztDataList = Lists.newArrayList();
        transaction.begin();


        try {
            /*event处理*/
            for (int i = 0; i < batchSize; i++) {
                event = channel.take();
                if (event != null) {//对事件进行处理
                    //event 的 body 为   "exec tail-event-$i , $i"
                    content = new String(event.getBody());
                    SztData sztData = new SztData();
                    if (content.contains(",")) {
                        String[] infos = content.split(",");
                        sztData.setCar_no(infos[0]);
                        sztData.setCard_no(infos[1]);
                        sztData.setClose_date(infos[2]);
                        sztData.setCompany_name(infos[3]);
                        sztData.setConn_mark(infos[4]);
                        sztData.setDeal_date(infos[5]);
                        sztData.setDeal_money(infos[6]);
                        sztData.setDeal_type(infos[7]);
                        sztData.setDeal_value(infos[8]);
                        sztData.setEqu_no(infos[9]);
                        sztData.setStation(infos[10]);
                    }
                    sztDataList.add(sztData);
                } else {
                    result = Status.BACKOFF;
                    break;
                }
            }

            /*jdbc提交*/
            if (sztDataList.size() > 0) {
                preparedStatement.clearBatch();
                for (SztData temp : sztDataList) {
                    preparedStatement.setString(1, temp.getCar_no());
                    preparedStatement.setString(2, temp.getCard_no());
                    preparedStatement.setString(3, temp.getClose_date());
                    preparedStatement.setString(4, temp.getCompany_name());
                    preparedStatement.setString(5, temp.getConn_mark());
                    preparedStatement.setString(6, temp.getDeal_date());
                    preparedStatement.setString(7, temp.getDeal_money());
                    preparedStatement.setString(8, temp.getDeal_type());
                    preparedStatement.setString(9, temp.getDeal_value());
                    preparedStatement.setString(10, temp.getEqu_no());
                    preparedStatement.setString(11, temp.getStation());
                    preparedStatement.addBatch();
                }
                preparedStatement.executeBatch();
                conn.commit();
            }
            transaction.commit();
        } catch (Exception e) {
            try {
                transaction.rollback();
            } catch (Exception e2) {
                LOG.error("Exception in rollback. Rollback might not have been.successful.", e2);
            }
            LOG.error("Failed to commit transaction.Transaction rolled back.", e);
            Throwables.propagate(e);
        } finally {
            transaction.close();
        }
        return result;
    }
}
